package cn.ltgodm.house.analysis;

import cn.ltgodm.house.util.SparkUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

import java.util.Arrays;
import java.util.List;

import static org.apache.spark.sql.functions.*;

/**
 * @author ltgodm
 * @date 2024-06-23 13:18:41
 */
@Service
@Slf4j
public class AnalysisService {
    @Resource
    private SparkUtil sparkUtil;

    //各地区单价分析
    public void districtPriceAnalysis() {
        Dataset<Row> dataFrame = sparkUtil.getDataFrame("/house/cleaned.csv");
        Dataset<Row> result = dataFrame.withColumn("单价", col("单价").cast("double"))
                .groupBy(col("区域位置").as("district"))
                .agg(round(avg("单价")).as("price"));
        sparkUtil.genMysqlTable(result, "district_price");
    }

    //装修单价分析
    public void decorationPriceAnalysis() {
        Dataset<Row> dataFrame = sparkUtil.getDataFrame("/house/cleaned.csv");
        Dataset<Row> result = dataFrame.withColumn("单价", col("单价").cast("double"))
                .groupBy(col("装修情况").as("decoration"))
                .agg(round(avg("单价")).as("price"));
        sparkUtil.genMysqlTable(result, "decoration_price");
    }

    //建筑类型单价分析
    public void buildingTypeCountAnalysis() {
        Dataset<Row> dataFrame = sparkUtil.getDataFrame("/house/cleaned.csv");

        double total = dataFrame.filter(col("建筑类型").notEqual("暂无数据"))
                .count();

        Dataset<Row> result = dataFrame
                .filter(col("建筑类型").notEqual("暂无数据"))
                .groupBy(col("建筑类型").as("building_type"))
                .count();


        List<Row> newRow = Arrays.asList(RowFactory.create("total", total));

        StructType schema = new StructType()
                .add("building_type", DataTypes.StringType)
                .add("count", DataTypes.DoubleType);

        Dataset<Row> newDF = sparkUtil.getSparkSession().createDataFrame(newRow, schema);

        result = result.union(newDF);

        sparkUtil.genMysqlTable(result, "building_type_count");
    }

    //房屋户型单价数量分析
    public void houseTypeCountAnalysis() {
        Dataset<Row> dataFrame = sparkUtil.getDataFrame("/house/cleaned.csv");

        Dataset<Row> result = dataFrame
                .groupBy(col("房屋户型").as("house_type"))
                .agg(round(avg("单价"), 2).as("price"), count(col("单价")).as("count")).orderBy("price");
        sparkUtil.genMysqlTable(result, "house_type_count");
    }

    public void areaAndAttentionAnalysis() {
        Dataset<Row> dataFrame = sparkUtil.getDataFrame("/house/cleaned.csv");
        double maxPrice = dataFrame.withColumn("总价", col("总价").cast("double")).agg(max("总价")).head().getDouble(0);
        double minPrice = dataFrame.withColumn("总价", col("总价").cast("double")).agg(min("总价")).head().getDouble(0);

        double step = (maxPrice - minPrice) / 10;

        // 计算套内面积价格
        Dataset<Row> result1 = dataFrame.withColumn("套内面积", col("套内面积").cast("double"))
                .withColumn("price_group", floor(col("总价").minus(minPrice).divide(step)))
                .groupBy(col("price_group"))
                .agg(round(avg("套内面积"), 1).as("inner_area_avg"))
                .na().fill(0)
                .withColumn("price_group", col("price_group").cast("double"))
                .withColumn("min_price", round(col("price_group").multiply(step).plus(minPrice), 2))
                .withColumn("max_price", round(col("price_group").plus(1).multiply(step).plus(minPrice), 2));

        // 计算建筑面积价格
        Dataset<Row> result2 = dataFrame.withColumn("建筑面积", col("建筑面积").cast("double"))
                .withColumn("price_group", floor(col("总价").minus(minPrice).divide(step)))
                .groupBy(col("price_group"))
                .agg(round(avg("建筑面积"), 1).as("building_area_avg"))
                .na().fill(0)
                .withColumn("price_group", col("price_group").cast("double"))
                .withColumn("min_price", round(col("price_group").multiply(step).plus(minPrice), 2))
                .withColumn("max_price", round(col("price_group").plus(1).multiply(step).plus(minPrice), 2));

        // 计算关注度
        Dataset<Row> result3 = dataFrame.withColumn("关注度", col("关注度").cast("double"))
                .withColumn("price_group", floor(col("总价").minus(minPrice).divide(step)))
                .groupBy(col("price_group"))
                .agg(round(avg("关注度"), 1).as("attention_avg"))
                .na().fill(0)
                .withColumn("price_group", col("price_group").cast("double"));

        //合并数据
        Dataset<Row> result = result1
                .join(result2, result1.col("price_group").equalTo(result2.col("price_group")), "outer")
                .join(result3, result1.col("price_group").equalTo(result3.col("price_group")), "outer")
                .select(result1.col("min_price"), result1.col("max_price"), result1.col("inner_area_avg"),
                        result2.col("building_area_avg"), result3.col("attention_avg")).na().fill(0);
        sparkUtil.genMysqlTable(result, "area_attention_price");
    }

    // 计算是否有电梯
    public void elevatorAnalysis() {
        Dataset<Row> dataFrame = sparkUtil.getDataFrame("/house/cleaned.csv");
        double maxPrice = dataFrame.withColumn("单价", col("单价").cast("double")).agg(max("单价")).head().getDouble(0);
        double step = (maxPrice - 0) / 10;

        // 计算配备电梯的房屋数量
        Dataset<Row> result4 = dataFrame.withColumn("单价", col("单价").cast("double"))
                .withColumn("price_group", floor(col("单价").minus(0).divide(step)))
                .filter(col("配备电梯").equalTo("有"))
                .groupBy(col("price_group"))
                .agg(count("单价").as("elevator_count"))
                .withColumn("price_group", col("price_group").cast("double"))
                .withColumn("min_price", round(col("price_group").multiply(step).plus(0).divide(10000), 2))
                .withColumn("max_price", round(col("price_group").plus(1).multiply(step).divide(10000), 2));

        // 计算不配备电梯的房屋数量
        Dataset<Row> result5 = dataFrame.withColumn("单价", col("单价").cast("double"))
                .withColumn("price_group", floor(col("单价").minus(0).divide(step)))
                .filter(col("配备电梯").equalTo("无"))
                .groupBy(col("price_group"))
                .agg(count("单价").as("no_elevator_count"))
                .withColumn("price_group", col("price_group").cast("double"))
                .withColumn("min_price", round(col("price_group").multiply(step).plus(0).divide(10000), 2))
                .withColumn("max_price", round(col("price_group").plus(1).multiply(step).divide(10000), 2));

        //合并数据
        Dataset<Row> result = result4
                .join(result5, result4.col("price_group").equalTo(result5.col("price_group")), "outer")
                .select( result4.col("min_price"), result4.col("max_price"), result4.col("elevator_count"), result5.col("no_elevator_count")).na().fill(0);
        sparkUtil.genMysqlTable(result, "elevator_count_price");
    }



    //电梯户比例价格分析
    public void elevatorHousePriceAnalysis() {
        Dataset<Row> dataFrame = sparkUtil.getDataFrame("/house/cleaned.csv");
        Dataset<Row> result = dataFrame.withColumn("单价", col("单价").cast("double"))
                .groupBy(col("梯户比例").as("elevator"))
                .agg(round(avg("单价"), 0).as("price"),count("单价").as("count"))
                .orderBy(col("price").desc(),col("count").desc() );
        sparkUtil.genMysqlTable(result, "house_elevator_price");
    }

    //所在楼层价格分析
    public void floorPriceAnalysis() {
        Dataset<Row> dataFrame = sparkUtil.getDataFrame("/house/cleaned.csv");
        Dataset<Row> result = dataFrame.withColumn("单价", col("单价").cast("double"))
                .groupBy(col("所在楼层").as("floor"))
                .agg(round(avg("单价"), 0).as("price"),count("单价").as("count"))
                .orderBy(col("price").desc(),col("count").desc() );
        sparkUtil.genMysqlTable(result, "floor_price");
    }
}
